package net.wlm.job;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;


public class DaMoPanJob {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 2. 输入表-input:映射到kafka消息队列
        createInputTable(tabEnv);
        // 3. 输出表-output:映射到clickhouse
        createOutputTable(tabEnv);
        // 4. 保存数据
        saveToDirs(tabEnv);

    }
    private static void createOutputTable(TableEnvironment tabEnv) {
        // 用户表
        tabEnv.executeSql("CREATE TABLE doris_user_profiles_sink (\n" +
                "    user_id STRING,\n" +
                "    register_time TIMESTAMP(3),\n" +
                "    gender INT,\n" +
                "    age_range STRING,\n" +
                "    province STRING,\n" +
                "    city STRING,\n" +
                "    is_parent BOOLEAN\n" +
                ")\n" +
                "WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = 'node102:8030',\n" +
                "    'table.identifier' = 'damopan.user_profiles',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'sink.batch.size' = '1000',\n" +
                "    'sink.properties.format' = 'json',\n" +
                "    'sink.properties.strip_outer_array' = 'true'\n" +
                ")");
        // 商品表
        tabEnv.executeSql("CREATE TABLE my_item_attributes (\n" +
                "    item_id STRING COMMENT '商品ID，主键',\n" +
                "    item_name STRING COMMENT '商品名称',\n" +
                "    category_id STRING  COMMENT '类目ID',\n" +
                "    category_path STRING COMMENT '完整类目路径',\n" +
                "    keywords STRING COMMENT '商品关键词',\n" +
                "    description STRING COMMENT '商品描述',\n" +
                "    price DOUBLE COMMENT '商品价格',\n" +
                "    is_parent BOOLEAN COMMENT '是否母婴类商品'\n" +
                ") WITH (\n" +
                "      'connector' = 'doris',\n" +
                "      'fenodes' = 'node102:8030',\n" +
                "      'table.identifier' = 'damopan.item_attributes',\n" +
                "      'username' = 'root',\n" +
                "      'password' = '123456',\n" +
                "      'sink.batch.size' = '1000',\n" +
                "      'sink.properties.format' = 'json',\n" +
                "      'sink.properties.strip_outer_array' = 'true'\n" +
                "      )");
        //  用户行为
        tabEnv.executeSql("CREATE TABLE my_user_behavior_logs (\n" +
                "         log_id STRING COMMENT '日志ID',\n" +
                "         user_id STRING COMMENT '用户ID',\n" +
                "         item_id STRING COMMENT '商品ID',\n" +
                "         behavior_type STRING COMMENT '行为类型:purchase/browse/collect/add_cart/search',\n" +
                "         behavior_time TIMESTAMP(3) COMMENT '行为时间',\n" +
                "         search_keyword STRING COMMENT '搜索关键词',\n" +
                "         session_id STRING COMMENT '会话ID',\n" +
                "         stay_duration BIGINT COMMENT '停留时长(ms)',\n" +
                "         page_url STRING  COMMENT '页面URL'\n" +
                ") WITH (\n" +
                "      'connector' = 'doris',\n" +
                "      'fenodes' = 'node102:8030',\n" +
                "      'table.identifier' = 'damopan.user_behavior_logs',\n" +
                "      'username' = 'root',\n" +
                "      'password' = '123456',\n" +
                "      'sink.batch.size' = '1000',\n" +
                "      'sink.properties.format' = 'json',\n" +
                "      'sink.properties.strip_outer_array' = 'true'\n" +
                "      )");
        // 订单交易
        tabEnv.executeSql("CREATE TABLE my_order_transactions (\n" +
                "          order_id varchar(20) comment '订单ID，主键',\n" +
                "          user_id varchar(20) comment '用户ID',\n" +
                "          item_id varchar(20) comment '商品ID',\n" +
                "          purchase_time TIMESTAMP(3) comment '购买时间',\n" +
                "          quantity INT comment '购买数量',\n" +
                "          amount DOUBLE comment   '订单金额',\n" +
                "          payment_method varchar(20) comment '支付方式',\n" +
                "          `status` INT comment '订单状态:1已完成/2已退款'\n" +
                ") WITH (\n" +
                "      'connector' = 'doris',\n" +
                "      'fenodes' = 'node102:8030',\n" +
                "      'table.identifier' = 'damopan.order_transactions',\n" +
                "      'username' = 'root',\n" +
                "      'password' = '123456',\n" +
                "      'sink.batch.size' = '1000',\n" +
                "      'sink.properties.format' = 'json',\n" +
                "      'sink.properties.strip_outer_array' = 'true'\n" +
                "      )");
    }
    private static void saveToDirs(TableEnvironment tabEnv) {
        // 插入语句
        // 用户
        tabEnv.executeSql("INSERT into doris_user_profiles_sink\n" +
                "SELECT \n" +
                "    user_id,\n" +
                "    register_time,\n" +
                "    gender,\n" +
                "    age_range,\n" +
                "    province,\n" +
                "    city,\n" +
                "    is_parent\n" +
                "FROM Kafka_user_profiles");
        // 商品
        tabEnv.executeSql("insert into my_item_attributes\n" +
                "select * from Kafka_item_attributes");
        // 用户行为
        tabEnv.executeSql("insert into my_user_behavior_logs \n" +
                "select * from Kafka_user_behavior_logs");
        // 订单交易
        tabEnv.executeSql("insert into my_order_transactions\n" +
                "select * from Kafka_order_transactions");
    }
    private static void createInputTable(TableEnvironment tabEnv) {
        // 用户基础信息
        tabEnv.executeSql("CREATE TABLE Kafka_user_profiles (\n" +
                "      user_id STRING,\n" +
                "      register_time TIMESTAMP(3),\n" +
                "      gender INT,\n" +
                "      age_range STRING,\n" +
                "      province STRING,\n" +
                "      city STRING,\n" +
                "      is_parent BOOLEAN)\n" +
                "    WITH (\n" +
                "      'connector' = 'kafka',\n" +
                "      'topic' = 'user_profile_topic',\n" +
                "      'scan.startup.mode' = 'earliest-offset',\n" +
                "      'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "      'properties.group.id' = 'testGroup',\n" +
                "      'format' = 'json',\n" +
                "      'json.fail-on-missing-field' = 'false',\n" +
                "      'json.ignore-parse-errors' = 'true'\n" +
                "      )");
        // 商品属性表
        tabEnv.executeSql("CREATE TABLE Kafka_item_attributes (\n" +
                "       item_id STRING,\n" +
                "       item_name STRING,\n" +
                "       category_id STRING,\n" +
                "       category_path STRING,\n" +
                "       keywords STRING,\n" +
                "       description STRING,\n" +
                "       price DOUBLE,\n" +
                "       is_parent BOOLEAN)\n" +
                "    WITH (\n" +
                "        'connector' = 'kafka',\n" +
                "        'topic' = 'item_attributes_topic',\n" +
                "        'scan.startup.mode' = 'earliest-offset',\n" +
                "        'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "        'properties.group.id' = 'testGroup',\n" +
                "        'format' = 'json',\n" +
                "        'json.fail-on-missing-field' = 'false',\n" +
                "        'json.ignore-parse-errors' = 'true'\n" +
                "        )");
        // 用户行为日志
        tabEnv.executeSql("CREATE TABLE Kafka_user_behavior_logs (\n" +
                "         log_id STRING,\n" +
                "         user_id STRING,\n" +
                "         item_id STRING,\n" +
                "         behavior_type STRING,\n" +
                "         behavior_time TIMESTAMP(3),\n" +
                "         search_keyword STRING,\n" +
                "         session_id STRING,\n" +
                "         stay_duration BIGINT,\n" +
                "         page_url STRING)\n" +
                "    WITH (\n" +
                "        'connector' = 'kafka',\n" +
                "        'topic' = 'user_behavior_topic',\n" +
                "        'scan.startup.mode' = 'earliest-offset',\n" +
                "        'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "        'properties.group.id' = 'testGroup',\n" +
                "        'format' = 'json',\n" +
                "        'json.fail-on-missing-field' = 'false',\n" +
                "        'json.ignore-parse-errors' = 'true'\n" +
                "        )");
        // 订单交易数据
        tabEnv.executeSql("CREATE TABLE Kafka_order_transactions (\n" +
                "         order_id STRING,\n" +
                "         user_id STRING,\n" +
                "         item_id STRING,\n" +
                "         purchase_time TIMESTAMP(3),\n" +
                "         quantity INT,\n" +
                "         amount DOUBLE,\n" +
                "         payment_method STRING,\n" +
                "         status INT)\n" +
                "    WITH (\n" +
                "        'connector' = 'kafka',\n" +
                "        'topic' = 'order_transaction_topic',\n" +
                "        'scan.startup.mode' = 'earliest-offset',\n" +
                "        'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "        'properties.group.id' = 'testGroup',\n" +
                "        'format' = 'json',\n" +
                "        'json.fail-on-missing-field' = 'false',\n" +
                "        'json.ignore-parse-errors' = 'true'\n" +
                "        )");

        // 查询数据
      // tabEnv.executeSql("select * from Kafka_user_profiles limit 5").print();
      // tabEnv.executeSql("select * from Kafka_item_attributes limit 5").print();
      // tabEnv.executeSql("select * from Kafka_user_behavior_logs limit 5").print();
      // tabEnv.executeSql("select * from Kafka_order_transactions limit 5").print();
    }
    private static TableEnvironment getTableEnv() {
        // a-环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // b-属性配置设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // c-返回对象
        return tabEnv;
    }
}